use crate::cluster::{HealthCheckResult, HealthCheckStatus, NodeStatus};
use crate::config::HealthCheckConfig;
use crate::events::EventSystem;
use crate::{Error, Result};

use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
use tokio::time::interval;
use tracing::{debug, error, info, warn};

/// Health checker that monitors node health
pub struct HealthChecker {
    config: HealthCheckConfig,
    event_system: Arc<EventSystem>,
    nodes: Arc<RwLock<HashMap<String, NodeHealthState>>>,
    shutdown_tx: Arc<RwLock<Option<tokio::sync::oneshot::Sender<()>>>>,
}

/// Health state for a single node
#[derive(Debug, Clone)]
struct NodeHealthState {
    address: String,
    status: NodeStatus,
    last_check: Instant,
    consecutive_failures: u32,
    consecutive_successes: u32,
    health_checks: Vec<HealthCheckResult>,
}

/// Health check types
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum HealthCheck {
    Connectivity,
    ResourceUsage,
    ApplicationHealth,
    Custom { name: String, endpoint: String },
}

/// Detailed health information for a node
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeHealth {
    pub node_id: String,
    pub status: NodeStatus,
    pub last_seen: chrono::DateTime<chrono::Utc>,
    pub checks: Vec<HealthCheckResult>,
    pub uptime_percentage: f64,
    pub response_time_avg: Duration,
}

impl HealthChecker {
    /// 创建新的健康检查器
    ///
    /// 保存配置与事件系统引用，初始化节点状态容器。
    pub async fn new(config: &HealthCheckConfig, event_system: Arc<EventSystem>) -> Result<Self> {
        info!("Initializing health checker");

        Ok(Self {
            config: config.clone(),
            event_system,
            nodes: Arc::new(RwLock::new(HashMap::new())),
            shutdown_tx: Arc::new(RwLock::new(None)),
        })
    }

    /// 启动健康检查后台循环
    ///
    /// 根据配置的 interval 周期性执行全量节点健康检查，支持 oneshot 关闭。
    pub async fn start(&self) -> Result<()> {
        info!("Starting health checker");

        let interval_duration = parse_duration(&self.config.interval)?;
        let (tx, rx) = tokio::sync::oneshot::channel();
        {
            let mut guard = self.shutdown_tx.write().await;
            *guard = Some(tx);
        }

        let nodes = self.nodes.clone();
        let config = self.config.clone();
        let event_system = self.event_system.clone();

        tokio::spawn(async move {
            let mut interval = interval(interval_duration);
            let mut shutdown = rx;

            loop {
                tokio::select! {
                    _ = interval.tick() => {
                        if let Err(e) = Self::run_health_checks(&nodes, &config, &event_system).await {
                            error!("Health check run failed: {}", e);
                        }
                    }
                    _ = &mut shutdown => {
                        info!("Health checker shutting down");
                        break;
                    }
                }
            }
        });

        Ok(())
    }

    /// 关闭健康检查循环（幂等）
    pub async fn shutdown(&self) -> Result<()> {
        info!("Shutting down health checker");

        if let Some(tx) = self.shutdown_tx.write().await.take() { let _ = tx.send(()); }

        Ok(())
    }

    /// 添加节点进入健康检查跟踪
    pub async fn add_node(&self, node_id: &str, address: &str) -> Result<()> {
        info!("Adding node {} to health checking", node_id);

        let state = NodeHealthState {
            address: address.to_string(),
            status: NodeStatus::Unknown,
            last_check: Instant::now(),
            consecutive_failures: 0,
            consecutive_successes: 0,
            health_checks: Vec::new(),
        };

        self.nodes.write().await.insert(node_id.to_string(), state);
        Ok(())
    }

    /// 从健康检查移除节点
    pub async fn remove_node(&self, node_id: &str) -> Result<()> {
        info!("Removing node {} from health checking", node_id);

        self.nodes.write().await.remove(node_id);
        Ok(())
    }

    /// 获取单节点健康详情（快照）
    pub async fn get_node_health(&self, node_id: &str) -> Result<NodeHealth> {
        let nodes = self.nodes.read().await;
        let state = nodes
            .get(node_id)
            .ok_or_else(|| Error::health_check(format!("Node {} not found", node_id)))?;

        Ok(NodeHealth {
            node_id: node_id.to_string(),
            status: state.status.clone(),
            last_seen: chrono::Utc::now(), // Convert from last_check
            checks: state.health_checks.clone(),
            uptime_percentage: 99.5, // Calculate from history
            response_time_avg: Duration::from_millis(50), // Calculate from history
        })
    }

    /// 判断节点是否处于 Healthy 状态
    pub async fn is_node_healthy(&self, node_id: &str) -> Result<bool> {
        let nodes = self.nodes.read().await;
        let state = nodes
            .get(node_id)
            .ok_or_else(|| Error::health_check(format!("Node {} not found", node_id)))?;

        Ok(matches!(state.status, NodeStatus::Healthy))
    }

    /// 触发指定节点立即健康检查（异步执行）
    pub async fn trigger_check(&self, node_id: &str) -> Result<()> {
        info!("Triggering immediate health check for node {}", node_id);

        let nodes = self.nodes.clone();
        let config = self.config.clone();
        let event_system = self.event_system.clone();

        // Run health check for specific node
        if let Err(e) = Self::check_node_health(node_id, &nodes, &config, &event_system).await {
            error!("Health check failed for node {}: {}", node_id, e);
        }

        Ok(())
    }

    /// 对全部已注册节点执行健康检查
    async fn run_health_checks(
        nodes: &Arc<RwLock<HashMap<String, NodeHealthState>>>,
        config: &HealthCheckConfig,
        event_system: &Arc<EventSystem>,
    ) -> Result<()> {
        debug!("Running health checks for all nodes");

        let node_ids: Vec<String> = {
            let nodes_guard = nodes.read().await;
            nodes_guard.keys().cloned().collect()
        };

        for node_id in node_ids {
            if let Err(e) = Self::check_node_health(&node_id, nodes, config, event_system).await {
                error!("Health check failed for node {}: {}", node_id, e);
            }
        }

        Ok(())
    }

    /// 执行单节点健康检查逻辑
    async fn check_node_health(
        node_id: &str,
        nodes: &Arc<RwLock<HashMap<String, NodeHealthState>>>,
        config: &HealthCheckConfig,
        event_system: &Arc<EventSystem>,
    ) -> Result<()> {
        let address = {
            let nodes_guard = nodes.read().await;
            let state = nodes_guard
                .get(node_id)
                .ok_or_else(|| Error::health_check(format!("Node {} not found", node_id)))?;
            state.address.clone()
        };

        debug!("Checking health for node {} at {}", node_id, address);

        let check_results = Self::perform_health_checks(&address, config).await?;
        let overall_healthy = check_results.iter().all(|check| {
            matches!(
                check.status,
                HealthCheckStatus::Pass | HealthCheckStatus::Warn
            )
        });

        // Update node state
        let status_changed = {
            let mut nodes_guard = nodes.write().await;
            if let Some(state) = nodes_guard.get_mut(node_id) {
                let old_status = state.status.clone();

                if overall_healthy {
                    state.consecutive_successes += 1;
                    state.consecutive_failures = 0;

                    if state.consecutive_successes >= config.recovery_threshold {
                        state.status = NodeStatus::Healthy;
                    }
                } else {
                    state.consecutive_failures += 1;
                    state.consecutive_successes = 0;

                    if state.consecutive_failures >= config.failure_threshold {
                        state.status = NodeStatus::Unhealthy;
                    }
                }

                state.last_check = Instant::now();
                state.health_checks = check_results;

                old_status != state.status
            } else {
                false
            }
        };

        // Emit event if status changed
        if status_changed {
            let new_status = {
                let nodes_guard = nodes.read().await;
                nodes_guard.get(node_id).unwrap().status.clone()
            };

            let status_str = match new_status {
                NodeStatus::Healthy => "healthy",
                NodeStatus::Unhealthy => "unhealthy",
                NodeStatus::Maintenance => "maintenance",
                NodeStatus::Unknown => "unknown",
            };

            event_system
                .emit_node_health_change(node_id, status_str)
                .await?;

            info!("Node {} health status changed to {}", node_id, status_str);
        }

        Ok(())
    }

    /// 针对配置列表执行具体检查项（connectivity / resource_usage / application_health）
    async fn perform_health_checks(
        address: &str,
        config: &HealthCheckConfig,
    ) -> Result<Vec<HealthCheckResult>> {
        let timeout = parse_duration(&config.timeout)?;
        let mut results = Vec::new();

        for check_name in &config.checks {
            let start = Instant::now();
            let result = match check_name.as_str() {
                "connectivity" => Self::check_connectivity(address, timeout).await,
                "resource_usage" => Self::check_resource_usage(address, timeout).await,
                "application_health" => Self::check_application_health(address, timeout).await,
                _ => {
                    warn!("Unknown health check: {}", check_name);
                    continue;
                }
            };

            let duration = start.elapsed();
            let (status, message) = match result {
                Ok(msg) => (HealthCheckStatus::Pass, msg),
                Err(e) => (HealthCheckStatus::Fail, Some(e.to_string())),
            };

            results.push(HealthCheckResult {
                name: check_name.clone(),
                status,
                duration,
                message,
            });
        }

        Ok(results)
    }

    /// 检查 HTTP 连通性（GET /health）
    async fn check_connectivity(address: &str, timeout: Duration) -> Result<Option<String>> {
        let client = reqwest::Client::builder()
            .timeout(timeout)
            .build()
            .map_err(|e| Error::health_check(format!("Failed to create HTTP client: {}", e)))?;

        let url = format!("http://{}/health", address);

        match client.get(&url).send().await {
            Ok(response) => {
                if response.status().is_success() {
                    Ok(Some(format!("HTTP {} from {}", response.status(), url)))
                } else {
                    Err(Error::health_check(format!(
                        "HTTP {} from {}",
                        response.status(),
                        url
                    )))
                }
            }
            Err(e) => Err(Error::health_check(format!("Connection failed: {}", e))),
        }
    }

    /// 检查资源使用（GET /metrics，当前仅判断成功/失败）
    async fn check_resource_usage(address: &str, timeout: Duration) -> Result<Option<String>> {
        let client = reqwest::Client::builder()
            .timeout(timeout)
            .build()
            .map_err(|e| Error::health_check(format!("Failed to create HTTP client: {}", e)))?;

        let url = format!("http://{}/metrics", address);

        match client.get(&url).send().await {
            Ok(response) => {
                if response.status().is_success() {
                    // Parse metrics and check thresholds
                    Ok(Some("Resource usage within limits".to_string()))
                } else {
                    Err(Error::health_check(format!(
                        "Metrics endpoint returned {}",
                        response.status()
                    )))
                }
            }
            Err(e) => Err(Error::health_check(format!("Failed to get metrics: {}", e))),
        }
    }

    /// 应用自定义健康检查（占位实现）
    async fn check_application_health(_address: &str, _timeout: Duration) -> Result<Option<String>> {
        // Implementation would depend on the specific application
        // For now, just return success
        Ok(Some("Application health check passed".to_string()))
    }
}

/// 解析带单位的时间字符串 (ns/us/ms/s/m/h)
fn parse_duration(s: &str) -> Result<Duration> {
    let s = s.trim();

    if s.is_empty() {
        return Err(Error::health_check("Empty duration string"));
    }

    let (number_part, unit_part) = if let Some(pos) = s.chars().position(|c| c.is_alphabetic()) {
        s.split_at(pos)
    } else {
        return Err(Error::health_check("Duration must have a unit"));
    };

    let number: f64 = number_part
        .parse()
        .map_err(|_| Error::health_check("Invalid number in duration"))?;

    let multiplier = match unit_part {
        "ns" => 1e-9,
        "us" | "μs" => 1e-6,
        "ms" => 1e-3,
        "s" => 1.0,
        "m" => 60.0,
        "h" => 3600.0,
        _ => return Err(Error::health_check("Invalid duration unit")),
    };

    let seconds = number * multiplier;
    Ok(Duration::from_secs_f64(seconds))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_health_checker_creation() {
        let config = HealthCheckConfig {
            interval: "10s".to_string(),
            timeout: "5s".to_string(),
            failure_threshold: 3,
            recovery_threshold: 2,
            checks: vec!["connectivity".to_string()],
        };

        let event_system = Arc::new(EventSystem::new());
        let health_checker = HealthChecker::new(&config, event_system).await;

        assert!(health_checker.is_ok());
    }

    #[test]
    fn test_parse_duration() {
        assert_eq!(parse_duration("5s").unwrap(), Duration::from_secs(5));
        assert_eq!(parse_duration("10m").unwrap(), Duration::from_secs(600));
        assert_eq!(parse_duration("500ms").unwrap(), Duration::from_millis(500));

        assert!(parse_duration("").is_err());
        assert!(parse_duration("5").is_err());
        assert!(parse_duration("5x").is_err());
    }
}
